package com.easyseata.tc.job;

import static com.easyseata.common.Constants.ASYNC_COMMITTING;
import static com.easyseata.common.Constants.RETRY_COMMITTING;
import static com.easyseata.common.Constants.RETRY_ROLLBACKING;
import static com.easyseata.common.Constants.TX_TIMEOUT_CHECK;
import static com.easyseata.common.Constants.UNDOLOG_DELETE;
import static com.easyseata.common.DefaultValues.DEFAULT_ASYNC_COMMITTING_RETRY_PERIOD;
import static com.easyseata.common.DefaultValues.DEFAULT_COMMITING_RETRY_PERIOD;
import static com.easyseata.common.DefaultValues.DEFAULT_MAX_COMMIT_RETRY_TIMEOUT;
import static com.easyseata.common.DefaultValues.DEFAULT_MAX_ROLLBACK_RETRY_TIMEOUT;
import static com.easyseata.common.DefaultValues.DEFAULT_ROLLBACKING_RETRY_PERIOD;
import static com.easyseata.common.DefaultValues.DEFAULT_ROLLBACK_RETRY_TIMEOUT_UNLOCK_ENABLE;

import java.util.Collection;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.easyseata.common.ConfigurationKeys;
import com.easyseata.common.thread.NamedThreadFactory;
import com.easyseata.common.util.CollectionUtils;
import com.easyseata.config.core.Configuration;
import com.easyseata.config.core.ConfigurationFactory;
import com.easyseata.core.exception.TransactionException;
import com.easyseata.core.model.GlobalStatus;
import com.easyseata.tc.coordinator.DefaultCore;
import com.easyseata.tc.session.GlobalSession;
import com.easyseata.tc.session.SessionHelper;
import com.easyseata.tc.session.SessionHolder;
import com.easyseata.tc.storage.file.session.SessionCondition;

public class CronTask {
	
	private static final int ALWAYS_RETRY_BOUNDARY = 0;
	
	private final DefaultCore core=DefaultCore.getInstance();
	
	protected static final Logger LOGGER = LoggerFactory.getLogger(CronTask.class);
	
    private static final long MAX_COMMIT_RETRY_TIMEOUT = ConfigurationFactory.getInstance().getLong(
            ConfigurationKeys.MAX_COMMIT_RETRY_TIMEOUT, DEFAULT_MAX_COMMIT_RETRY_TIMEOUT);

    private static final long MAX_ROLLBACK_RETRY_TIMEOUT = ConfigurationFactory.getInstance().getLong(
            ConfigurationKeys.MAX_ROLLBACK_RETRY_TIMEOUT, DEFAULT_MAX_ROLLBACK_RETRY_TIMEOUT);
    

    private static final boolean ROLLBACK_RETRY_TIMEOUT_UNLOCK_ENABLE = ConfigurationFactory.getInstance().getBoolean(
            ConfigurationKeys.ROLLBACK_RETRY_TIMEOUT_UNLOCK_ENABLE, DEFAULT_ROLLBACK_RETRY_TIMEOUT_UNLOCK_ENABLE);

    private final GlobalStatus[] rollbackingStatuses = new GlobalStatus[] {GlobalStatus.TimeoutRollbacking,
	        GlobalStatus.TimeoutRollbackRetrying, GlobalStatus.RollbackRetrying, GlobalStatus.Rollbacking};
    
    private final GlobalStatus[] retryCommittingStatuses = new GlobalStatus[] {GlobalStatus.Committing, GlobalStatus.CommitRetrying, GlobalStatus.Committed};
	 /**
     * The constant CONFIG.
     */
    protected static final Configuration CONFIG = ConfigurationFactory.getInstance();
	 /**
     * The constant COMMITTING_RETRY_PERIOD.
     */
    protected static final long COMMITTING_RETRY_PERIOD = CONFIG.getLong(ConfigurationKeys.COMMITING_RETRY_PERIOD,
            DEFAULT_COMMITING_RETRY_PERIOD);

    /**
     * The constant ASYNC_COMMITTING_RETRY_PERIOD.
     */
    protected static final long ASYNC_COMMITTING_RETRY_PERIOD = CONFIG.getLong(
            ConfigurationKeys.ASYNC_COMMITING_RETRY_PERIOD, DEFAULT_ASYNC_COMMITTING_RETRY_PERIOD);

    /**
     * The constant ROLLBACKING_RETRY_PERIOD.
     */
    protected static final long ROLLBACKING_RETRY_PERIOD = CONFIG.getLong(ConfigurationKeys.ROLLBACKING_RETRY_PERIOD,
            DEFAULT_ROLLBACKING_RETRY_PERIOD);

	private final ScheduledThreadPoolExecutor retryRollbacking = new ScheduledThreadPoolExecutor(1,
			new NamedThreadFactory(RETRY_ROLLBACKING, 1));

	private final ScheduledThreadPoolExecutor retryCommitting = new ScheduledThreadPoolExecutor(1,
			new NamedThreadFactory(RETRY_COMMITTING, 1));

	private final ScheduledThreadPoolExecutor asyncCommitting = new ScheduledThreadPoolExecutor(1,
			new NamedThreadFactory(ASYNC_COMMITTING, 1));

	private final ScheduledThreadPoolExecutor timeoutCheck = new ScheduledThreadPoolExecutor(1,
			new NamedThreadFactory(TX_TIMEOUT_CHECK, 1));

	private final ScheduledThreadPoolExecutor undoLogDelete = new ScheduledThreadPoolExecutor(1,
			new NamedThreadFactory(UNDOLOG_DELETE, 1));
	
	
	
	
	public CronTask() {
		super();
		this.init();
	}


	/**
     * Init.
     */
    public void init() {
        retryRollbacking.scheduleAtFixedRate(
            () -> SessionHolder.distributedLockAndExecute(RETRY_ROLLBACKING, this::handleRetryRollbacking), 0,
            ROLLBACKING_RETRY_PERIOD, TimeUnit.MILLISECONDS);

        retryCommitting.scheduleAtFixedRate(
            () -> SessionHolder.distributedLockAndExecute(RETRY_COMMITTING, this::handleRetryCommitting), 0,
            COMMITTING_RETRY_PERIOD, TimeUnit.MILLISECONDS);

    }
    
    
    /**
     * Handle retry rollbacking.
     */
    protected void handleRetryRollbacking() {
        SessionCondition sessionCondition = new SessionCondition(rollbackingStatuses);
        sessionCondition.setLazyLoadBranch(true);
        Collection<GlobalSession> rollbackingSessions =
            SessionHolder.getRootSessionManager().findGlobalSessions(sessionCondition);
        if (CollectionUtils.isEmpty(rollbackingSessions)) {
            return;
        }
        long now = System.currentTimeMillis();
        SessionHelper.forEach(rollbackingSessions, rollbackingSession -> {
            try {
                // prevent repeated rollback
                if (rollbackingSession.getStatus() == GlobalStatus.Rollbacking
                    && !rollbackingSession.isDeadSession()) {
                    // The function of this 'return' is 'continue'.
                    return;
                }
                if (isRetryTimeout(now, MAX_ROLLBACK_RETRY_TIMEOUT, rollbackingSession.getBeginTime())) {
                    if (ROLLBACK_RETRY_TIMEOUT_UNLOCK_ENABLE) {
                        rollbackingSession.clean();
                    }

                    SessionHelper.endRollbackFailed(rollbackingSession, true, true);

                    //The function of this 'return' is 'continue'.
                    return;
                }
                core.doGlobalRollback(rollbackingSession, true);
            } catch (TransactionException ex) {
                LOGGER.error("Failed to retry rollbacking [{}] {} {}", rollbackingSession.getXid(), ex.getCode(), ex.getMessage());
            }
        });
    }
    
    private boolean isRetryTimeout(long now, long timeout, long beginTime) {
        return timeout >= ALWAYS_RETRY_BOUNDARY && now - beginTime > timeout;
    }
    
    /**
     * Handle retry committing.
     */
    protected void handleRetryCommitting() {
        SessionCondition retryCommittingSessionCondition = new SessionCondition(retryCommittingStatuses);
        retryCommittingSessionCondition.setLazyLoadBranch(true);
        Collection<GlobalSession> committingSessions =
            SessionHolder.getRootSessionManager().findGlobalSessions(retryCommittingSessionCondition);
        if (CollectionUtils.isEmpty(committingSessions)) {
            return;
        }
        long now = System.currentTimeMillis();
        SessionHelper.forEach(committingSessions, committingSession -> {
            try {
                // prevent repeated commit
                if ((GlobalStatus.Committing.equals(committingSession.getStatus())
                        || GlobalStatus.Committed.equals(committingSession.getStatus()))
                        && !committingSession.isDeadSession()) {
                    // The function of this 'return' is 'continue'.
                    return;
                }
                if (isRetryTimeout(now, MAX_COMMIT_RETRY_TIMEOUT, committingSession.getBeginTime())) {

                    // commit retry timeout event
                    SessionHelper.endCommitFailed(committingSession, true, true);

                    //The function of this 'return' is 'continue'.
                    return;
                }
                if (GlobalStatus.Committed.equals(committingSession.getStatus())
                    && committingSession.getBranchSessions().isEmpty()) {
                    SessionHelper.endCommitted(committingSession,true);
                }
                core.doGlobalCommit(committingSession, true);
            } catch (TransactionException ex) {
                LOGGER.error("Failed to retry committing [{}] {} {}", committingSession.getXid(), ex.getCode(), ex.getMessage());
            }
        });
    }
}
